package com.itcj.dmp.tags

import com.itcj.dmp.casename.tagcase
import groovy.sql.DataSet
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}

import scala.collection.{immutable, mutable}

object MakeTags {

  import com.itcj.dmp.utils.KuduHelper._

  def makeTodayTag(sparkSession: SparkSession): Dataset[tagcase] = {
    import com.itcj.dmp.casename.TableName._
    //获取今日数据表
    val odsdf: DataFrame = sparkSession.readKuduTable(todayTableName).get
    val areadf: DataFrame = sparkSession.readKuduTable(areaTableName).get
    //join两个表
    val joindf: DataFrame = odsdf.join(areadf, odsdf.col("geohash") === areadf.col("geohash"), "left")
          .where(areadf.col("area").isNotNull)
    joindf.select("area").show()
    //通过遍历join表的每一行数据，得到tagcase
    import sparkSession.implicits._
    val tagds: Dataset[tagcase] = joindf.map(row => {
      val ids: Map[String, String] = getIds(row)
      val mainid: String = getMainId(ids)
      val tags: Map[String, Double] = getTags(row)
      tagcase(mainid, ids, tags)
    })
    tagds
  }


  //获取ips
  def getIds(row: Row): Map[String, String] = {
    val keyList: List[String] = List("imei", "imeimd5", "imeisha1", "mac", "macmd5", "macsha1", "openudid",
      "openudidmd5", "openudidsha1", "idfa", "idfamd5", "idfasha1")
    val ids1: immutable.Seq[(String, String)] = keyList.map(key => {
      val value: String = row.getAs[String](key)
      (key, value)
    })
    val ids2 = ids1.toMap
    ids2
  }

  //获取主ip
  def getMainId(ids: Map[String, String]): String = {
    val keyList: List[String] = List("imei", "imeimd5", "imeisha1", "mac", "macmd5", "macsha1", "openudid",
      "openudidmd5", "openudidsha1", "idfa", "idfamd5", "idfasha1")
    for (key <- keyList) {
      val ms: Option[String] = ids.get(key)
      val value1 = ms.get
      if (ms.isDefined && value1 != "") {
        return value1
      }
    }
    return ""
  }

  def getTags(row: Row): Map[String, Double] = {
    var areamap: mutable.Map[String, Double] = mutable.Map[String, Double]()
    areamap += ("AD_" + row.getAs[String]("adspacetype") -> 1.0)
    areamap += ("CH_" + row.getAs[String]("channelid") -> 1)
    row.getAs[String]("keywords").split(",").foreach(kw => {
      areamap += ("KW_" + kw -> 1)
    })

    areamap += ("PN_" + row.getAs[String]("region") -> 1)
    areamap += ("CN_" + row.getAs[String]("city") -> 1)
    areamap += ("SEX_" + row.getAs[String]("sex") -> 1)
    areamap += ("AGE_" + row.getAs[String]("age") -> 1)

    row.getAs[String]("area").split(",").map((_ -> 1.0)).foreach(areamap += _)
    areamap.toMap
  }

}
